package com.liip.app.ftp;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.pool2.ObjectPool;

import lombok.extern.slf4j.Slf4j;

/**
 * 自定义实现ftp连接池
 * @Auther: luolin
 * @Date: 2020-10-14 4:06 PM
 * @Description:
 */
@Slf4j
public class FTPClientPool implements ObjectPool<FTPClient>{
	
	private static final int DEFAULT_POOL_SIZE = 10;
	 
	  public BlockingQueue<FTPClient> blockingQueue;
	 
	  private FTPClientFactory factory;
	 
	  public FTPClientPool(FTPClientFactory factory) throws Exception {
	    this(DEFAULT_POOL_SIZE, factory);
	  }
	 
	  public FTPClientPool(int poolSize, FTPClientFactory factory) throws Exception {
	    this.factory = factory;
	    this.blockingQueue = new ArrayBlockingQueue<FTPClient>(poolSize);
	    initPool(poolSize);
	  }
	 
	  /**
	   * 初始化连接池
	   * @param maxPoolSize
	   *         最大连接数
	   * @throws Exception
	   */
	  private void initPool(int maxPoolSize) throws Exception {
	    int count = 0;
	    while(count < maxPoolSize) {
	      this.addObject();
	      count++;
	    }
	  }
	 
	  /**
	   * 从连接池中获取对象
	   */
	  @Override
	  public FTPClient borrowObject() throws Exception {
	    FTPClient client = blockingQueue.take();
	    if(client == null) {
	      client = factory.create();
	    } else if(!factory.validateObject(factory.wrap(client))) {
	      invalidateObject(client);
	      client = factory.create();
	    }
	    return client;
	  }
	 
	  /**
	   * 返还一个对象(链接)
	   */
	  @Override
	  public void returnObject(FTPClient client) throws Exception {
	    if ((client != null) && !blockingQueue.offer(client,2,TimeUnit.MINUTES)) {
	      try {
	        factory.destroyObject(factory.wrap(client));
	      } catch (Exception e) {
	        throw e;
	      }
	    }
	  }
	 
	  /**
	   * 移除无效的对象(FTP客户端)
	   */
	  @Override
	  public void invalidateObject(FTPClient client) throws Exception {
	    blockingQueue.remove(client);
	  }
	 
	  /**
	   * 增加一个新的链接，超时失效
	   */
	  @Override
	  public void addObject() throws Exception {
	    blockingQueue.offer(factory.create(), 2, TimeUnit.MINUTES);
	  }
	 
	  /**
	   * 重新连接
	   */
	  public FTPClient reconnect() throws Exception {
	    return factory.create();
	  }
	 
	  /**
	   * 获取空闲链接数(这里暂不实现)
	   */
	  @Override
	  public int getNumIdle() {
	    return blockingQueue.size();
	  }
	 
	  /**
	   * 获取正在被使用的链接数
	   */
	  @Override
	  public int getNumActive() {
	    return DEFAULT_POOL_SIZE - getNumIdle();
	  }
	 
	  @Override
	  public void clear() throws Exception {
	 
	  }
	 
	  /**
	   * 关闭连接池
	   */
	  @Override
	  public void close() {
	    try {
	      while(blockingQueue.iterator().hasNext()) {
	        FTPClient client = blockingQueue.take();
	        factory.destroyObject(factory.wrap(client));
	      }
	    } catch(Exception e) {
	      log.error("close ftp client pool failed...{}", e);
	    }
	  }
	 
	  /**
	   * 增加一个新的链接，超时失效
	   */
	  public void addObject(FTPClient ftpClient) throws Exception {
	    blockingQueue.put(ftpClient);
	  }

}
